package network;

import com.google.common.base.Throwables;
import com.sun.jdi.request.InvalidRequestStateException;
import operation.JodisReceive;
import server.JodisConfig;
import utils.Closeables;
import utils.Logger;

import javax.naming.ConfigurationException;
import java.io.IOException;
import java.net.Socket;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * 用单线程去处理的读写事件，在redis6.0中改用多线程方式，我这里的接受和分发逻辑是两个线程
 * **/
public class Processor extends AbstractAccepter{
    private final Logger logger = Logger.getLogger(getClass());

    private BlockingQueue<SocketChannel> newConnections = null;
    private RequestHandler requestHandler;
    private int maxRequestSize;

    /**
     * @param jodisConfig
     * @param requestHandler
     */
    public Processor(JodisConfig jodisConfig, RequestHandler requestHandler) throws ConfigurationException {
        this.newConnections = new ArrayBlockingQueue<>(jodisConfig.getMaxConcurrency());
        this.requestHandler = requestHandler;
        this.maxRequestSize = jodisConfig.getMaxRequestSize();
    }


    //newConnections都是已建立连接的事件
    public void accept(SocketChannel socketChannel) {
        newConnections.add(socketChannel);
        getSelector().wakeup();
    }

    @Override
    public void run() {
        startupComplete();
        while (isRunning()){
            try{
                configureNewConnections();
                final Selector selector = getSelector();
                //监听是否有发生的事件
                int ready = selector.select(500);
                if (ready <= 0) {
                    continue;
                }

                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while(iterator.hasNext() && isRunning()){
                    SelectionKey key = null;
                    try{
                        key = iterator.next();
                        iterator.remove();
                        if(key.isReadable()){
                            read(key);
                        }else if(key.isWritable()){
                            write(key);
                        }else if(!key.isValid()){
                            close(key);
                        }else{
                            throw new IllegalStateException("Unrecognized key state for processor thread.");
                        }
                    }catch (InvalidRequestStateException e){
                        Socket socket = channelFor(key).socket();
                        logger.info("Closing socket connection to {}:{} due to invalid request: {}",
                                socket.getInetAddress(),
                                socket.getPort(),
                                e.getMessage());
                        close(key);
                    } catch (Throwable t) {
                        Socket socket = channelFor(key).socket();
                        final String msg = "Closing socket for {}:{} because of error {}";
                        logger.info(msg, socket.getInetAddress(), socket.getPort(), Throwables.getStackTraceAsString(t));
                        close(key);
                    }
                }
            }catch (Exception e){
                logger.error(e);
            }
        }

        logger.info("Closing selector while shutting down");
        closeSelector();
        shutdownComplete();
    }

    private SocketChannel channelFor(SelectionKey key) {
        return (SocketChannel) key.channel();
    }

    private void close(SelectionKey key){
        SocketChannel channel = (SocketChannel) key.channel();
        logger.info("Closing connection from {}", channel.socket().getRemoteSocketAddress());
        Closeables.closeQuietly(channel.socket());
        Closeables.closeQuietly(channel);
        key.attach(null);
        key.cancel();
    }

    private void write(SelectionKey key) throws IOException{
        Send response = (Send) key.attachment();
        SocketChannel socketChannel = channelFor(key);
        response.write(socketChannel);
        if (response.completed()) {
            key.attach(null);
            //确定了下一次调用某个选择器的选择方法时，将测试哪类操作的准备就绪信息
            key.interestOps(SelectionKey.OP_READ);
        } else {
            key.interestOps(SelectionKey.OP_WRITE);
            getSelector().wakeup();
        }
    }

    private void read(SelectionKey key){
        SocketChannel socketChannel = channelFor(key);
        Receive receive = null;
        if(key.attachment() == null){
            receive = new JodisReceive(maxRequestSize);
            //每一个key绑定一个数据对象
            key.attach(receive);
        }else{
            receive = (Receive) key.attachment();
        }

        int read = receive.read(socketChannel);
        if(read < 0){
            close(key);
        }else if(receive.complete()){
            //数据读取完毕，需要去对读出来的指令做响应
            Send maybeResponse = handle(receive);
            key.attach(null);
            // if there is a response, send it, otherwise do nothing
            if (maybeResponse != null) {
                key.attach(maybeResponse);
                key.interestOps(SelectionKey.OP_WRITE);
            }
        }else{
            // more reading to be done
            key.interestOps(SelectionKey.OP_READ);
            getSelector().wakeup();
            logger.info("reading request not been done {}", receive);
        }
    }

    private Send handle(Receive request) {
        return requestHandler.handle(request);
    }

    //批量将已连接事件注册到selector
    private void configureNewConnections() throws ClosedChannelException {
        while (newConnections.size() > 0) {
            SocketChannel channel = newConnections.poll();
            logger.info("Listening to new connection from {}", channel.socket().getRemoteSocketAddress());
            channel.register(getSelector(), SelectionKey.OP_READ);
        }
    }

}
